#include "epollserver.h"

#include <sys/time.h>
#include <sys/resource.h>

using namespace std;

void FdTimeOutThread::run()
{
    while(!m_exitflag)
    {
        /// 主动关闭超时连接
        std::set<int>   timeoutFds;
        m_server.getTimeoutFds(timeoutFds);
        
        int closed = 0;
        for(set<int>::iterator it = timeoutFds.begin();
                it != timeoutFds.end(); ++it)
        {
            if (*it < 0)
                continue;
            
            ::close(*it);
            ++closed;
        }
        
        m_server.reduceMaxEvent(closed);
        if (closed == 0)
        {
            m_event.wait(1000);
        }
    }
}

EpollServer::EpollServer(const SocketAddress& address,
        EpollDispatcher& dispather,
        bool reuseAddress,
        EPOLL_MODE mode,
        int backlog)
    :m_dispatcher(dispather)
    ,m_mode(mode)
{
    Socket::bind(address, reuseAddress);
    Socket::listen(backlog);
    m_epollFd   = 0;
    m_exitflag  = false;
    init();
}

EpollServer::EpollServer(unsigned int port,
        EpollDispatcher& dispather,
        bool reuseAddress,
        EPOLL_MODE mode,
        int backlog)
    :m_dispatcher(dispather)
    ,m_mode(mode)
{
    SocketAddress address("0.0.0.0", port);
    Socket::bind(address, reuseAddress);
    Socket::listen(backlog);
    m_epollFd = 0;
    m_exitflag =false;
    init();
}

EpollServer& EpollServer::operator = (const Socket& socket)
{
    Socket::operator = (socket);
    return *this;
}

EpollServer::~EpollServer()
{
    /// 资源类的都需要手动关闭
    stopWait();
    ::close(m_epollFd);
    
    if (m_timeoutTarget)
        m_timeoutTarget->stop();
    
    m_timeoutThread.join();
    if (m_timeoutTarget)
    {
        delete m_timeoutTarget;
        m_timeoutTarget = NULL;
    }
}

void EpollServer::init(unsigned long maxopenfd)
{
    struct rlimit rt;
    /* 设置每个进程允许打开的最大文件数 */
    rt.rlim_max = rt.rlim_cur = maxopenfd;
    setrlimit(RLIMIT_NOFILE, &rt);
    
    // epoll init
     /*
     * Since Linux 2.6.8, the size argument is ignored, but must be greater than zero
     */
    m_epollFd   = epoll_create(maxopenfd);
    struct epoll_event ev;
    ev.data.fd  = sockfd();
    if (m_mode == EPOLLMODE_ET)
    {
        ev.events   = EPOLLIN | EPOLLET;
        Socket::setBlocking(sockfd(),false);
    }
    else
    {
        ev.events   = EPOLLIN;
    }
    
    epoll_ctl(m_epollFd,EPOLL_CTL_ADD,sockfd(),&ev);
    m_event = 0;
    m_timeoutTarget = new FdTimeOutThread(*this,m_epollFd);
    m_timeoutThread.start(*m_timeoutTarget);
}

void EpollServer::closingFd(int timeoutFd)
{
    ScopedMutex   lock(&m_timeoutFdsMutex);
    m_timeoutFds.insert(timeoutFd);
    m_timeoutTarget->m_event.set();
}

void EpollServer::getTimeoutFds(std::set<int>& timeoutFds)
{
    ScopedMutex   lock(&m_timeoutFdsMutex);
    timeoutFds = m_timeoutFds;
    m_timeoutFds.clear();
}

void EpollServer::wait()
{
    int nfds                    = 0;
    int etimeout                = 2;   //mili second
    
    struct epoll_event          ev;
    ev.data.fd                  = 0;
    if (m_mode == EPOLLMODE_ET)
    {
        ev.events               = EPOLLIN|EPOLLRDHUP|EPOLLET|EPOLLERR;
    }
    else
    {
        ev.events               = EPOLLIN|EPOLLRDHUP|EPOLLERR;
    }
    m_event                     = ev.events;
    std::vector<epoll_event>    events;
    unsigned int defaultsize    = 1024;
    events.resize(defaultsize);
    m_maxevent                  = 1;    // include m_epollFd

    while(!m_exitflag)
    {
        nfds = epoll_wait(m_epollFd,&events[0],events.size(),etimeout);
        int fd_added = 0;
        for(int i = 0; i < nfds; ++i)
        {
            if(events[i].data.fd == sockfd())
            {
                int tmpfd = 0;
                while( (tmpfd = acceptConnection())  > 0)
                {
                    ev.data.fd   = tmpfd;
                    if (m_mode == EPOLLMODE_ET)
                    {
                        Socket::setBlocking(tmpfd,false);
                        //Socket::setNoDelay(tmpfd,true);
                    }

                    if( 0 == epoll_ctl(m_epollFd,EPOLL_CTL_ADD,tmpfd,&ev))
                    {
                        ++fd_added;
                    }
                }
            }
            // 客户端 close 会同时触发 EPOLLIN 和 EPOLLRDHUP，用EPOLLIN只表示可读
            else if (events[i].events&EPOLLRDHUP )
            {
                m_dispatcher.dequeue(events[i].data.fd);
                closingFd(events[i].data.fd);
            }
            else if( events[i].events&EPOLLIN )
            {
                m_dispatcher.enqueue(events[i].data.fd);
            }
            // events[i].events&EPOLLERR
            // events[i].events&EPOLLHUP
            else
            {
                m_dispatcher.dequeue(events[i].data.fd);
                closingFd(events[i].data.fd);
            }
        }
        addMaxEvent(fd_added);
        //events.clear();
        if (m_maxevent < defaultsize)
            m_maxevent = defaultsize;
        else if ( events.size() < m_maxevent)
            events.resize(m_maxevent+10000);
        else if (events.size() - m_maxevent > 10000)
            events.resize(events.size() - 10000);
    }
}

int EpollServer::acceptConnection()
{
    return Socket::acceptConnection();
}







